package com.deepglint.sink;

import com.deepglint.beans.SensorReading;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

import java.util.Properties;

/**
 * @author mj
 * @version 1.0
 * @date 2021-11-16 21:58
 */
public class SinkTest_Redis {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        String inputPath = "C:\\Users\\马军\\Desktop\\Idea-workspace\\flink\\src\\main\\resources\\source.txt";
        DataStream<String> inputStream = env.readTextFile(inputPath);

        // 转换成SensorReading类型
        DataStream<SensorReading> streamMap = inputStream.map(value -> {
            String[] split = value.split(",");
            return new SensorReading(split[0], split[1], new Long(split[2]), new Double(split[3]));
        });

        // 定义连接参数
        FlinkJedisPoolConfig config = new FlinkJedisPoolConfig.Builder().
                setHost("192.168.150.128")
                .setPort(6379)
                .build();
        // 将数据存入redis
        streamMap.addSink(new RedisSink<>(config, new MyRedisMapper()));

        env.execute();
    }

    // 自定义redismapper
    static class MyRedisMapper implements RedisMapper<SensorReading> {

        // 定义保存数据到redis的命令，存成hash表(hset)，sensor_temp:id temperature
        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.HSET, "sensor_temp");
        }

        @Override
        public String getKeyFromData(SensorReading data) {
            return data.getId();
        }

        @Override
        public String getValueFromData(SensorReading data) {
            return data.getTemperature().toString();
        }
    }
}


